package org.xwb.springcloud.messagetype.deadletter;

import com.alibaba.fastjson2.util.DateUtils;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.xwb.springcloud.constant.MqDeadLetterConstant;

import java.io.IOException;
import java.util.Date;

/**
 * @author 10564
 */
@Component
public class DeadLetterBusinessAndConsumer {
    private static final Logger log = LoggerFactory.getLogger(DeadLetterBusinessAndConsumer.class);

    /**
     * @param msg     消息内容
     * @param channel 通道对象
     * @param tag     消息的tag
     */
    @RabbitListener(queues = MqDeadLetterConstant.BUSINESS_QUEUE)
    public void receiveAckBusinessQueueMessage(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
        try {
            log.info("\nd消费者eadLetter接收消息:【{}】\n", msg);
            if ("normal".equals(msg)) {
                log.info("\n已被正常消费 normal 【{}】,【{}】\n", msg, DateUtils.format(new Date()));
                //todo 消息正常处理，从队列中删除掉当前消息
                channel.basicAck(tag, false);
            } else if ("deadLetter".equals(msg)) {
                log.info("\n模拟消费异常,需要进入死信队列中 deadLetter 【{}】,【{}】\n", msg, DateUtils.format(new Date()));
                //channel.basicReject(tag, false);也可以使用
                channel.basicNack(tag, false,false);
            } else {
                throw new Exception(msg);
            }
        } catch (Exception e) {
            log.info("\n消费者处理消息异常:【{}】\n", e.getMessage());
            //todo true 允许加入队列， false 拒绝加入队列 ，进入私信队列中
            channel.basicReject(tag, false);
        }
    }

    /**
     * 死信队列消费者
     *
     * @param msg     消息
     * @param channel 通道对象
     * @param tag     消息的tag
     */
    @RabbitListener(queues = MqDeadLetterConstant.DEAD_LETTER_QUEUE)
    public void receiveAckDlqQueue(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        try {
            // 处理死信队列消息
            log.error("\n死信消费者接收消息:【{}】\n", msg);
            //todo true 允许加入队列， false 拒绝加入队列
            channel.basicAck(tag, false);
        } catch (Exception e) {
            log.info("\n死信消费者消息异常:【{}】\n", e.getMessage());
            try {
                channel.basicNack(tag, false, true);
            } catch (IOException ex) {
                throw new RuntimeException(ex);
            }
        }
    }
}
